package cn.hyxy.hadoop;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Demo20_SecondarySortMR extends Configured implements Tool {

	@Override
	public int run(String[] args) throws Exception {
		if (args.length != 2) {
			System.out.println("usage : in out...");
			return -1;
		}
		Configuration config = getConf();
		FileSystem fs = FileSystem.get(config);
		Path path = new Path(args[1]);
		if (fs.exists(path)) {
			fs.delete(path, true);
		}

		Job job = Job.getInstance(config, "SecondarySort");
		job.setJarByClass(getClass());

		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Temperature.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setNumReduceTasks(2);
		job.setPartitionerClass(YearPartitoner.class);
		
		job.setGroupingComparatorClass(YearGroupingComparator.class);
		
		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, path);

		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		int code = ToolRunner.run(new Demo20_SecondarySortMR(), args);
		System.exit(code);
	}
	
	//5：开发Reducer,此时只有一行数据，但是key和value里都是一个集合，只需要取key的第一条数据即可
	/**                  key                                  ,value
	 *  第1个文件： [2001  {2001/02/14:41},{2001/02/13:39},...],[null,null,...]
	 *  第2个文件： [2002  {2002/01/11:34},{2002/02/13:30},...],[null,null,...]
	 */
	public static class MyReducer extends Reducer<Temperature,NullWritable,Text,NullWritable>{
		@Override
		protected void reduce(Temperature key, Iterable<NullWritable> value,
				Reducer<Temperature, NullWritable, Text, NullWritable>.Context context)
				throws IOException, InterruptedException {		
			/*
			 * key1:2001	2001/01/12	16
				key1:2001	2001/01/13	18
				key1:2001	2001/01/11	23
				......
			 */
			/*for (NullWritable nullWritable : value) {
				System.out.println("key1:"+key);  //这是输出全部
			}*/
			
			context.write(new Text(key.toString()), NullWritable.get());
		}
	}
	
	//4：分组---开发分组的规则
	/**
	 * 如果连续（注意，一定连续）的两条或多条记录满足同组（即compare方法返回0）的条件，
	 * 即使key不相同，他们的value也会进入同一个values,执行一个reduce方法。
	 *                       key                                  ,value
	 *  第1个文件： [2001  {2001/02/14:41},{2001/02/13:39},...],[null,null,...]
	 *  第2个文件： [2002  {2002/01/11:34},{2002/02/13:30},...],[null,null,...]
	 */
	public static class YearGroupingComparator extends WritableComparator{
		public YearGroupingComparator() {//必须创建构造方法
			super(Temperature.class,true); //后边的参数代表：是否进行实例化（不能省！！！
		}
		
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			Temperature t1=(Temperature) a;
			Temperature t2=(Temperature) b;
			return t1.getYear().compareTo(t2.getYear());
		}
	}	
	
	//3；声明Partitioner，按“年”分区，输出2个文件,
	 /* 第1个：
	  * 2001	2001/01/12	16
		2001	2001/01/13	18
		......
		
		第2个：
		2002	2002/01/12	12
		2002	2002/01/13	18
		.....
	  */
	public static class YearPartitoner extends Partitioner<Temperature, NullWritable>{
		@Override
		public int getPartition(Temperature temperature, NullWritable nullWritable, int numPartitioners) {
			int year=Integer.parseInt(temperature.getYear());
			int p=year%numPartitioners;
			return p;
		}
	}
	
	//2:声明Mapper读取数据
	public static class MyMapper extends Mapper<LongWritable,Text,Temperature,NullWritable>{
		private Temperature temperature;
		@Override
		protected void map(LongWritable key, Text value,
				Mapper<LongWritable, Text, Temperature, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String[] strs=value.toString().split("\\s+");
			System.out.println(strs[0]+"\t"+strs[1]);
			String year=strs[0].split("/")[0];
			System.out.println(year);
			temperature=new Temperature(year, strs[0], Integer.parseInt(strs[1]));
			context.write(temperature, NullWritable.get());
		}
	}
	
	//1:声明Temperature的JavaBean
	public static class Temperature implements WritableComparable<Temperature>{
		private String year;
		private String date;
		private int temp;	
		
		@Override
		public String toString() {
			return year + "\t" + date + "\t" + temp;
		}
		
		public Temperature() {
		}
		
		public Temperature(String year, String date, int temp) {
			this.year = year;
			this.date = date;
			this.temp = temp;
		}

		@Override
		public int compareTo(Temperature t) {
			int comp=this.year.compareTo(t.getYear());
			if (comp==0) {
				comp=-(this.temp-t.getTemp());
			}
			return comp;
		}
		@Override
		public void readFields(DataInput in) throws IOException {
			year=in.readUTF();
			date=in.readUTF();
			temp=in.readInt();
		}

		@Override
		public void write(DataOutput out) throws IOException {
			out.writeUTF(year);
			out.writeUTF(date);
			out.writeInt(temp);
		}


		public String getYear() {
			return year;
		}

		public void setYear(String year) {
			this.year = year;
		}

		public String getDate() {
			return date;
		}

		public void setDate(String date) {
			this.date = date;
		}

		public int getTemp() {
			return temp;
		}

		public void setTemp(int temp) {
			this.temp = temp;
		}
		
	}
}
